package drds.binlog.prometheus.impl;

import drds.binlog.protocol.entry_protocol_proto.EntryType;
import drds.binlog.protocol.entry_protocol_proto.EventType;
import drds.binlog.sink.AbstractEventDownStreamHandler;
import drds.binlog.store.model.Event;

import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Chuanyi Li
 */
public class PrometheusEventDownStreamHandler extends AbstractEventDownStreamHandler<List<Event>>
{

    private final AtomicLong latestExecuteTime = new AtomicLong(System.currentTimeMillis());
    private final AtomicLong transactionCounter = new AtomicLong(0L);

    @Override
    public List<Event> before(List<Event> events)
    {
        long localExecTime = 0L;
        if (events != null && !events.isEmpty())
        {
            for (Event e : events)
            {
                EntryType type = e.getEntryType();
                if (type == null) continue;
                switch (type)
                {
                    case TRANSACTIONBEGIN:
                    {
                        long exec = e.getExecuteDateTimeLong();
                        if (exec > 0) localExecTime = exec;
                        break;
                    }
                    case ROWDATA:
                    {
                        long exec = e.getExecuteDateTimeLong();
                        if (exec > 0) localExecTime = exec;
                        break;
                    }
                    case TRANSACTIONEND:
                    {
                        long exec = e.getExecuteDateTimeLong();
                        if (exec > 0) localExecTime = exec;
                        transactionCounter.incrementAndGet();
                        break;
                    }
                    case HEARTBEAT:
                        EventType eventType = e.getEventType();
                        if (eventType == EventType.MHEARTBEAT)
                        {
                            localExecTime = System.currentTimeMillis();
                        }
                        break;
                    default:
                        break;
                }
            }
            if (localExecTime > 0)
            {
                latestExecuteTime.lazySet(localExecTime);
            }
        }
        return events;
    }

    @Override
    public void start()
    {

        super.start();
    }

    @Override
    public void stop()
    {
        super.stop();
    }

    public AtomicLong getLatestExecuteTime()
    {
        return latestExecuteTime;
    }

    public AtomicLong getTransactionCounter()
    {
        return transactionCounter;
    }

}
